package org.example.flume.sink;

import org.apache.flume.*;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * a1.sinks.k1.type = org.example.flume.sink.MySink
 */
public class MySink extends AbstractSink {
    private Status status;
    private final Logger logger = LoggerFactory.getLogger(MySink.class);
    @Override
    public Status process() throws EventDeliveryException {
        Channel c = getChannel();
        Transaction transaction = c.getTransaction();
        transaction.begin();
        try {
            Event event;
            while (true) {
                event = c.take();
                if (event == null) {
                    TimeUnit.SECONDS.sleep(1);
                    continue;
                }
                break;
            }
            storeSomeData(event);
            transaction.commit();
            status = Status.READY;
        } catch (ChannelException e) {
            e.printStackTrace();
            status = Status.BACKOFF;
        } catch (InterruptedException e) {
            e.printStackTrace();
            status = Status.BACKOFF;
        } finally {
            transaction.close();
        }

        return status;
    }

    private void storeSomeData(Event event) {
        logger.info(new String(event.getBody()));
    }
}
